package com.mpgame.caller;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.google.protobuf.Message;
import com.mpgame.common.connection.GrpcConnectionPool;
import com.mpgame.common.network.GrpcClient;
import com.mpgame.common.proto.*;
import com.mpgame.common.utils.Log;
import com.mpgame.connection.ChannelRegistry;

import io.grpc.stub.StreamObserver;
import io.netty.channel.ChannelHandlerContext;

public class GameServiceCaller {
    private volatile StreamObserver<StreamPacket> streamObs;
    private final GrpcConnectionPool pool;
    private final Empty empty;
    private int getAccountId(ChannelHandlerContext ctx){
        return Integer.parseInt(ChannelRegistry.getAccountId(ctx.channel()));
    }
    private void sendSuccessRsp(ChannelHandlerContext ctx, CmdId cmd, com.google.protobuf.Message payload){
        ctx.writeAndFlush(GamePacket.newBuilder()
            .setCmd(cmd)
            .setData(payload.toByteString())
            .build());
    }
    
    private void sendErrRsp(ChannelHandlerContext ctx, CmdId cmd, String err){
        ctx.writeAndFlush(GamePacket.newBuilder()
            .setCmd(cmd)
            .setData(com.google.protobuf.ByteString.copyFromUtf8(err == null ? "" : err))
            .build());
    }
    public GameServiceCaller(GrpcConnectionPool pool) {
        this.pool = pool;
        empty = Empty.newBuilder().build();
        establishStream();
    }
    private void establishStream(){
        try{
            GrpcClient cli = pool.borrowConnection();
            if(cli == null || !cli.isHealthy()){
                Log.err("无法获取GameServer连接，流连接建立失败！");
                return;
            }
            StreamObserver<StreamPacket> rspObs = new StreamObserver<StreamPacket>() {
                @Override
                public void onNext(StreamPacket rsp) {
                    handleStreamRsp(rsp);
                }
                @Override
                public void onError(Throwable t) {
                    Log.err("GameServer 流连接错误: " + t.getMessage());
                    streamObs = null;
                    // 尝试重连
                    scheduleReconnect();
                }
                @Override
                public void onCompleted() {
                    Log.info("GameServer 流连接完成");
                    streamObs = null;
                }
            };
            streamObs = GameServerServiceGrpc.newStub(cli.getChannel())
                .gameStream(rspObs);
            Log.info("成功建立到GameServer的流连接！");
            //TODO: 启动心跳

        }catch(Exception e){
            Log.err("建立GameServer流连接失败：" + e.getMessage());
            scheduleReconnect();
        }
    }
    private final Object sendLock = new Object();// 发送锁
    private void sendStreamPkt(CmdId msgType, Message msg){
        synchronized(sendLock){
            if(streamObs != null){
                streamObs.onNext(
                    StreamPacket.newBuilder()
                    .setCmd(msgType)
                    .setEc(ErrorCode.SUCCESS)
                    .setData(msg.toByteString())
                    .build());
            }else{
                Log.err("与gameServer双向流连接未建立！");
            }
        }
    }
    private void handleStreamRsp(StreamPacket rsp){
        if(rsp.getEc() == ErrorCode.SUCCESS){
            if(rsp.getTargetAccountIdCount() == 0) return;
            GamePacket pkt = GamePacket.newBuilder().setCmd(rsp.getCmd()).setData(rsp.getData()).build();
            Set<String> targets = new HashSet<>(rsp.getTargetAccountIdCount());
            for(int id : rsp.getTargetAccountIdList()) targets.add(String.valueOf(id));
            ChannelRegistry.multicast(pkt, targets);
        }else{
            Log.err("流连接返回错误：" + String.valueOf(rsp.getData()));
        }
    }
    /**
     * 尝试重连双向流
     */
    private void scheduleReconnect() {
        ScheduledExecutorService reconnectScheduler = Executors.newSingleThreadScheduledExecutor();
        reconnectScheduler.schedule(() -> {
            Log.info("尝试重新建立 GameServer 流连接");
            establishStream();
        }, 5, TimeUnit.SECONDS);
    }
    public boolean handleGameInitData(ChannelHandlerContext ctx, GamePacket pkt) {
        try{
            GameInitDataReq req = GameInitDataReq.parseFrom(pkt.getData());
            req = GameInitDataReq.newBuilder(req).setAccountId(getAccountId(ctx)).build();
    
            GrpcClient cli = pool.borrowConnection();
            if(cli == null || !cli.isHealthy()){
                sendErrRsp(ctx, CmdId.GAME_INITDATA_RSP, "服务暂时不可用");
                return true;
            }
    
            GameServerServiceGrpc.GameServerServiceStub stub =
                GameServerServiceGrpc.newStub(cli.getChannel()).withDeadlineAfter(10, TimeUnit.SECONDS);
    
            stub.getInitData(req, new io.grpc.stub.StreamObserver<GameInitDataRsp>(){
                @Override
                public void onNext(GameInitDataRsp rsp){
                    try{
                        if(rsp.getErrMsg() != null && !rsp.getErrMsg().isEmpty()){
                            sendErrRsp(ctx, CmdId.GAME_INITDATA_RSP, rsp.getErrMsg());
                            return;
                        }
                        sendSuccessRsp(ctx, CmdId.GAME_INITDATA_RSP, rsp);
                    }finally{
                        // no-op
                    }
                }
                @Override
                public void onError(Throwable t){
                    sendErrRsp(ctx, CmdId.GAME_INITDATA_RSP, "服务调用失败");
                    pool.returnConnection(cli);
                }
                @Override
                public void onCompleted(){
                    pool.returnConnection(cli);
                }
            });
            return true; // 立即返回，不阻塞IO线程
        }catch(Exception e){
            sendErrRsp(ctx, CmdId.GAME_INITDATA_RSP, "请求解析异常");
            return true;
        }
    }
    public boolean handleGameMoveTo(ChannelHandlerContext ctx, GamePacket pkt) {
        try{
            GameMoveToReq req = GameMoveToReq.parseFrom(pkt.getData());
            req = GameMoveToReq.newBuilder(req).setAccountId(getAccountId(ctx)).build();
            sendStreamPkt(CmdId.GAME_MOVE_TO_REQ, req);
            return true;
        }catch(Exception e){
            Log.err("处理玩家移动异常：" + e.getMessage());
            return false;
        }
    }
    public boolean handleGameChat(ChannelHandlerContext ctx, GamePacket pkt) {
        try{
            GameChatReq req = GameChatReq.parseFrom(pkt.getData());
            req = GameChatReq.newBuilder(req).setAccountId(getAccountId(ctx)).build();
            sendStreamPkt(CmdId.GAME_CHAT_REQ, req);
            return true;
        }catch(Exception e){
            Log.err("处理玩家聊天异常：" + e.getMessage());
            return false;
        }
    }
    public boolean handleGameSwitchScene(ChannelHandlerContext ctx, GamePacket pkt){
        try{
            GameSwitchSceneReq req = GameSwitchSceneReq.parseFrom(pkt.getData());
            req = GameSwitchSceneReq.newBuilder(req).setAccountId(getAccountId(ctx)).build();
            sendStreamPkt(CmdId.GAME_SWITCH_SCENE_REQ, req);
            return true;
        }catch(Exception e){
            Log.err("处理玩家切换场景异常：" + e.getMessage());
            return false;
        }
    }
}
